package com.nz2004.smsplatform.interfaces.mq;

import com.nz2004.smsplatform.common.constants.CacheConstants;
import com.nz2004.smsplatform.common.constants.RabbitMQConstants;
import com.nz2004.smsplatform.common.index.StandardReport;
import com.nz2004.smsplatform.interfaces.config.RabbitMQDelayedConfig;
import com.nz2004.smsplatform.interfaces.service.api.CacheService;
import com.nz2004.smsplatform.interfaces.util.JsonUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;

import java.io.IOException;
import java.util.Map;

/**
 * @author ReJoice
 */
@Component
@Slf4j
public class SmsReportListener {

    @Autowired
    private CacheService cacheService;

    @Autowired
    private RestTemplate restTemplate;

    private final String SUCCESS = "SUCCESS";

    private final int[] DELAYED_TIME = {0,5000,10000,30000,60000};

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @StreamListener(RabbitMQConstants.TOPIC_PUSH_SMS_REPORT)
    public void listen(StandardReport report,
                       @Header(name = AmqpHeaders.CHANNEL)Channel channel,
                       @Header(name = AmqpHeaders.DELIVERY_TAG)Long deliveryTag) throws IOException {
        log.info("【接口模块】 接收到状态报告，并推送给客户！！");
        //1. 获取当前客户的id
        long clientID = report.getClientID();
        //2. 调用缓存模块,查询客户信息.
        Map map = cacheService.hMGet(CacheConstants.CACHE_PREFIX_CLIENT + clientID);
        //3. 判断isreturnstatus是否为1,如果为1,代表需要返回
        Integer isReturnStatus =  Integer.valueOf((String) map.get("isreturnstatus"));
        //4. 如果为1,查询用户的回调地址receivestatusurl
        if(isReturnStatus == 1) {
            //5. 使用RestTemplate发送请求到receivestatusurl地址,并且携带参数,获取返回结果,返回结果为SUCCESS,代表客户正常的接收到了请求
            String receiveStatusUrl = (String) map.get("receivestatusurl");
            report.setReceiveStatusUrl(receiveStatusUrl);
            // 第一次发送
            sendRequestToCustomer(report);
        }
        //5. 手动ack
        channel.basicAck(deliveryTag,false);
    }

    private HttpEntity httpEntity(StandardReport report){
        //1. 封装请求头
        HttpHeaders header = new HttpHeaders();
        header.setContentType(MediaType.APPLICATION_JSON_UTF8);

        //2. 封装请求参数
        String parameters = JsonUtils.object2Json(report);

        //3. 将请求头和请求参数,封装到HttpEntity
        HttpEntity httpEntity = new HttpEntity(parameters,header);

        //4. 返回
        return httpEntity;
    }



    // 重新发送
    @RabbitListener(queues = {RabbitMQDelayedConfig.DELAYED_QUEUE})
    public void delayedListen(StandardReport report, Channel channel, Message message) throws IOException {
        log.info("【接口模块】 第{}次重新发送消息!!!",report.getSendCount());
        // 如果成功,没啥事,如果失败,发送消息到延迟队列
        sendRequestToCustomer(report);
        // 手动ack
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

    private void sendRequestToCustomer(StandardReport report) {
        try {
            int sendCount = report.getSendCount(); // 1
            report.setSendCount(++sendCount); // 2
            HttpEntity httpEntity = this.httpEntity(report); // 封装参数
            String resp = restTemplate.postForObject(report.getReceiveStatusUrl(), httpEntity, String.class); // 发送请求
            if (StringUtils.isEmpty(resp) || !resp.equals(SUCCESS)) {
                log.info("【接口模块】 发送请求失败!!");
                if(report.getSendCount() == 5){
                    return ;
                }
                // 客户接收请求失败!!!
                rabbitTemplate.convertAndSend(RabbitMQDelayedConfig.DELAYED_EXCHANGE, "", report, new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        message.getMessageProperties().setDelay(DELAYED_TIME[report.getSendCount()]);
                        return message;
                    }
                });
            }
        } catch (RestClientException e) {
            log.info("【接口模块】 发送请求失败!!");
            if(report.getSendCount() == 5){
                return ;
            }
            // 客户接收请求失败!!!
            rabbitTemplate.convertAndSend(RabbitMQDelayedConfig.DELAYED_EXCHANGE, "", report, new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setDelay(DELAYED_TIME[report.getSendCount()]);
                    return message;
                }
            });
        }
    }

}
